package collect

import (
	"sync"
	"time"

	"github.com/micro-plat/lib4go/logger"
	"github.com/micro-plat/lib4go/types"
	"github.com/prometheus/client_golang/prometheus"
)

// KafkaCollector 定义了一个用于收集kafka日志数据的结构体
type KafkaCollector struct {
	urlErrorCounter *prometheus.CounterVec
	log             logger.ILogger
	mu              sync.Mutex // 用于保护文件读取的并发访问
	durationHours   int
	fields          []string
}

// NewKafkaCollector 创建一个新的NginxCollector实例
func NewKafkaCollector(durationHours int, fields ...string) *KafkaCollector {
	urlErrorCounter := prometheus.NewCounterVec(
		prometheus.CounterOpts{
			Name: "url_error_total",
			Help: "URL错误总数",
		},
		fields,
	)

	nc := &KafkaCollector{
		urlErrorCounter: urlErrorCounter,
		durationHours:   durationHours,
		fields:          fields,
		log:             logger.New("kafka-exporter"),
	}
	go nc.startReset()
	return nc

}

func (c *KafkaCollector) Collectors() []prometheus.Collector {
	collects := make([]prometheus.Collector, 0, 3)
	collects = append(collects, c.urlErrorCounter)
	return collects
}
func (c *KafkaCollector) reset() {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.urlErrorCounter.Reset()
}

// Collect 收集指标
func (c *KafkaCollector) Collect(lineChan <-chan types.XMap) {
	go func() {
		for {
			select {
			case line, ok := <-lineChan:
				if !ok {
					return
				}
				c.setEntry(line)

			}
		}

	}()
}
func (c *KafkaCollector) startReset() {
	c.log.Infof("间隔%d小时重置一次计数器", c.durationHours)
	duration := time.Duration(c.durationHours) * time.Hour //按指定的时间间隔重置
	if c.durationHours == 24 {                             // 24小时则每天凌晨重置
		now := time.Now()
		nextMidnight := time.Date(now.Year(), now.Month(), now.Day()+1, 0, 0, 0, 0, now.Location())
		duration = nextMidnight.Sub(now)
	}
	timer := time.NewTimer(duration)
	for range timer.C {
		c.reset()
		timer.Reset(time.Duration(c.durationHours) * time.Hour) // 重置为下一个时间
		c.log.Infof("重置计数器完成，下次执行%dh后", c.durationHours)
	}
}

func (c *KafkaCollector) setEntry(url types.XMap) {
	c.mu.Lock()
	defer c.mu.Unlock()
	//获取多个字段的值
	values := make([]string, 0, len(c.fields))
	for _, field := range c.fields {
		values = append(values, url.GetString(field))
	}
	c.urlErrorCounter.WithLabelValues(values...).Inc()
}
